package com.bawei.flink.demo;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SessionWindowTimeGapExtractor;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;

/**
 *
 * 输入
 * 1631838600000,5000,hello
 * 1631838602000,5000,hello
 * 1631838605000,3000,hello
 * 1631838609000,5000,hello
 * 1631838610000,15000,hello
 * 1631838621000,5000,hello
 * 1631838628000,10000,hello
 * 1631838639000,10000,hello
 * 1631838645000,10000,hello
 *
 *
 * 输出
 * 2021-09-17 08:30:00  ---->  2021-09-17 08:30:08    hello,hello,hello,
 * 2021-09-17 08:30:09  ---->  2021-09-17 08:30:26    hello,hello,hello,
 *
 *
 */
public class DynamicSessionWindowDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> lines = env.socketTextStream("cdh1", 9999);

        lines
                .map(new MapFunction<String, Tuple3<Long,Long,String>>() {
            @Override
            public Tuple3<Long, Long, String> map(String value) throws Exception {
                String[] arr = value.split(",");
                return Tuple3.of(Long.valueOf(arr[0]),Long.valueOf(arr[1]),arr[2]);
            }
        })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<Long, Long, String>>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<Long, Long, String>>() {
            @Override
            public long extractTimestamp(Tuple3<Long, Long, String> element, long recordTimestamp) {
                return element.f0;
            }
        }))
                .windowAll(EventTimeSessionWindows.withDynamicGap(new SessionWindowTimeGapExtractor<Tuple3<Long, Long, String>>() {
            @Override
            public long extract(Tuple3<Long, Long, String> element) {
                return element.f1;
            }
        }))
                .process(new ProcessAllWindowFunction<Tuple3<Long, Long, String>, String, TimeWindow>() {

                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    @Override
                    public void process(Context context, Iterable<Tuple3<Long, Long, String>> elements, Collector<String> out) throws Exception {
                        String windowStart = sdf.format(context.window().getStart());
                        String windowEnd = sdf.format(context.window().getEnd());
                        StringBuffer sb = new StringBuffer();
                        for(Tuple3<Long,Long,String> element : elements) {
                            sb.append(element.f2+",");
                        }

                        String outStr = windowStart + "  ---->  " + windowEnd + "    " + sb.toString();
                        out.collect(outStr);
                    }
        })
                .print();
        env.execute();
    }
}
